package com.pingan.haofang.searchcloud.batch.index;

import com.pingan.haofang.searchcloud.batch.common.Batch;
import com.pingan.haofang.searchcloud.common.configuration.AmqpConfiguration;
import com.pingan.haofang.searchcloud.common.constants.Status;
import com.pingan.haofang.searchcloud.index.constants.IndexBuildStatus;
import com.pingan.haofang.searchcloud.index.constants.IndexBuildType;
import com.pingan.haofang.searchcloud.index.constants.IndexDataSourceType;
import com.pingan.haofang.searchcloud.index.dto.IndexBuildProgressDTO;
import com.pingan.haofang.searchcloud.index.service.IndexBuildProgressService;
import com.pingan.haofang.searchcloud.indexdatasource.dto.IndexDatasourceDTO;
import com.pingan.haofang.searchcloud.indexdatasource.dto.ScheduleTaskDto;
import com.pingan.haofang.searchcloud.indexdatasource.facade.IndexDatasourceScheduleFacade;
import com.pingan.haofang.searchcloud.quartz.constants.QuartzConstants;
import com.pingan.haofang.searchcloud.quartz.dto.ArgsDto;
import com.pingan.haofang.searchcloud.quartz.dto.ScheduleTask;
import com.pingan.haofang.searchcloud.quartz.dto.TaskQueryDto;
import com.pingan.haofang.searchcloud.quartz.service.IndexBuildScheduleExecution;
import com.pingan.haofang.searchcloud.quartz.service.ScheduleTaskManager;
import com.rabbitmq.client.Channel;
import org.apache.commons.collections.CollectionUtils;
import org.quartz.SchedulerException;
import org.quartz.TriggerKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.listener.MessageListenerContainer;
import org.springframework.amqp.rabbit.listener.MethodRabbitListenerEndpoint;
import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistry;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.jmx.export.annotation.ManagedAttribute;
import org.springframework.jmx.export.annotation.ManagedResource;
import org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory;
import org.springframework.scheduling.annotation.Scheduled;

import javax.annotation.Resource;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * 索引增量定时任务刷新
 *
 * @author LUYI374
 * @date 2017年5月8日
 * @since 1.0.0
 */
@Batch(value = "indexIncScheduleTask", permanent = true)
@ConditionalOnProperty("indexIncScheduleTask")
@ManagedResource
public class IndexIncScheduleTaskRunner {

    static Logger LOG = LoggerFactory.getLogger(IndexIncScheduleTaskRunner.class);

    @Resource
    private RabbitListenerEndpointRegistry rabbitListenerEndpointRegistry;

    @Resource(name = "containerFactory")
    private SimpleRabbitListenerContainerFactory containerFactory;

    @Autowired
    private IndexIncMessageListener messageListener;

    @Autowired
    private RabbitAdmin rabbitAdmin;

    @Autowired
    private IndexDatasourceScheduleFacade indexDatasourceScheduleFacade;

    @Autowired
    private IndexBuildProgressService indexBuildProgressService;

    @Autowired
    private ScheduleTaskManager scheduleTaskManager;

    @Autowired
    private IndexBuildScheduleExecution indexBuildScheduleExecution;

    /**
     * 任务刷新调度
     */
    @Deprecated
    @Scheduled(initialDelay = 5000, fixedDelay = 10000)
    public void refresh() throws SchedulerException {
        LOG.debug("start execute inc message listener reload");
        Method method = null;
        try {
            method = IndexIncMessageListener.class.getMethod("onMessage", Message.class, Channel.class);
        } catch (NoSuchMethodException | SecurityException e) {
            LOG.error("error get method for name: onMessage", e);
            throw new ExceptionInInitializerError(e);
        }

        List<ScheduleTaskDto> scheduleTaskDtos = indexDatasourceScheduleFacade.getAllEffectiveTasks(new TaskQueryDto());
        for (ScheduleTaskDto dto : scheduleTaskDtos) {

            List<IndexDatasourceDTO> dataSourceList = dto.getIndexDatasourceDTOs();

            Map<String, IndexBuildProgressDTO> datasourceMap = new HashMap<>(indexBuildProgressService
                    .query(dto.getProjectCode(), dto.getIndexName(), IndexBuildType.INC, IndexBuildStatus.RUNNING));

            boolean flag = false;
            for (IndexDatasourceDTO datasource : dataSourceList) {
                if (datasource.getIncDatasourceType() != null
                        && datasource.getIncDatasourceType() != IndexDataSourceType.NOT_SURPPORT.getValue()) {
                    flag = true;
                    if (datasourceMap.containsKey(datasource.getName())) {
                        datasourceMap.remove(datasource.getName());
                    }
                }
            }

            String name = dto.getProjectCode() + "_" + dto.getIndexName();

            if (flag) {

                // MQ配置声明
                rabbitAdmin.declareQueue(new Queue(name, true));
                rabbitAdmin.declareBinding(new Binding(name, Binding.DestinationType.QUEUE,
                        AmqpConfiguration.INDEX_BUILD_INC_EXCHANGE, name, null));

                MessageListenerContainer container = rabbitListenerEndpointRegistry.getListenerContainer(name);
                if (container == null) {

                    LOG.info("start inc progress, name:{}, type:{}", name, IndexBuildType.INC.desc);

                    MethodRabbitListenerEndpoint endpoint = new MethodRabbitListenerEndpoint();

                    endpoint.setMethod(method);
                    endpoint.setBean(messageListener);
                    endpoint.setMessageHandlerMethodFactory(new DefaultMessageHandlerMethodFactory());
                    endpoint.setId(name);
                    endpoint.setQueueNames(name);

                    rabbitListenerEndpointRegistry.registerListenerContainer(endpoint, containerFactory,
                            true);

                    for (String datasourceName : datasourceMap.keySet()) {
                        indexBuildProgressService.stopProgress(dto.getProjectCode(), dto.getIndexName(), datasourceName,
                                IndexBuildType.INC);
                    }
                }
            } else {
                MessageListenerContainer container = rabbitListenerEndpointRegistry.getListenerContainer(name);
                if (container != null) {
                    LOG.info("stop inc progress, name:{}, type:{}", name, IndexBuildType.INC.desc);
                    container.stop();

                    indexBuildProgressService.stopProgress(dto.getProjectCode(), dto.getIndexName(), null,
                            IndexBuildType.INC);
                }
            }

            List<ScheduleTask> result = new ArrayList<>();
            // 增量合并处理
            for (IndexDatasourceDTO datasourceDTO : dto.getIndexDatasourceDTOs()) {
                if (IndexDataSourceType.NOT_SURPPORT.getValue() != datasourceDTO.getIncDatasourceType()
                        && null != datasourceDTO.getIncDatasourceRpc()
                        && Status.NORMAL.getValue() == datasourceDTO.getIncDatasourceRpc().getBufferStatus()) {

                    LOG.info("schedule task manager registr inc index datasource {}", dto.toString());

                    ScheduleTask scheduleIncTask = new ScheduleTask();
                    scheduleIncTask.setGroup(dto.getProjectCode());
                    scheduleIncTask.setName(dto.getIndexName() + "_inc");
                    scheduleIncTask.setDesc(dto.getProjectCode() + "_" + dto.getIndexName());
                    scheduleIncTask.setTarget(QuartzConstants.TARGET);

                    ArgsDto argsDto = new ArgsDto();
                    argsDto.setProgressId(0L);
                    argsDto.setIndexDatasourceList(dto.getIndexDatasourceDTOs());
                    scheduleIncTask.setArgs(argsDto);

                    scheduleIncTask.setMethod(QuartzConstants.INC_METHOD);
                    // 每秒执行一次
                    scheduleIncTask.setCron("*/1 * * * * ?");
                    result.add(scheduleIncTask);
                    scheduleTaskManager.register(scheduleIncTask);
                }
            }
        }

        // 删除不用的调度
        Set<TriggerKey> newKeys = scheduleTaskDtos.stream()
                .map(item -> TriggerKey.triggerKey(item.getIndexName() + "_inc", item.getProjectCode()))
                .collect(Collectors.toSet());

        Collection deletingKeys =
                CollectionUtils.removeAll(scheduleTaskManager.getAll().keySet(), newKeys);
        for (Object key : deletingKeys) {
            TriggerKey triggerKey = (TriggerKey) key;
            scheduleTaskManager.delete(triggerKey.getGroup(), triggerKey.getName());
        }

    }

    /**
     * 任务进程状态刷新
     */
    @Scheduled(initialDelay = 30000, fixedDelay = 30000)
    public void progressSync() {
        LOG.debug("start execute inc progress sync");
        try {
            indexBuildProgressService.flushRunningProgress();
        } catch (Exception e) {
            LOG.error("sync index inc running progress failed", e);
        }
    }

    /**
     * jmx监控
     *
     * @return
     */
    @ManagedAttribute
    public RabbitListenerEndpointRegistry getRabbitListenerEndpointRegistry() {
        return rabbitListenerEndpointRegistry;
    }
}
